#include	"a_global_vars.h"

//------------------------------------------------------------------------------
void get_learner_request(int sd, short event, void *arg) {

	UNUSED_ARG(arg);
	UNUSED_ARG(event);

	gap request;
	request.start = 0;
	request.end = 0;
	int i = 0;
	int status;
	printf("Waiting to read the data from the socket: %d....\n", sd);
	status = recv(sd, &request, sizeof(gap), MSG_WAITALL);
	printf("socket: %d [%d-%d]	stat: %d\n", sd, request.start, request.end, status);
	assert(status == sizeof(gap));

	for (i = request.start; i <= request.end; i++) {
		//		memcpy(&temp, &data[(i % INTERNAL_BUFFER_SIZE)], sizeof(inst));
		status = send(sd, &data[(i % INTERNAL_BUFFER_SIZE)], sizeof(inst), MSG_WAITALL);
		assert(status == sizeof(inst));
	}
}
//------------------------------------------------------------------------------
void look4highest(int start, int end, int index) {
	int i;

	for (i = start; i < end; i++) {
		if (data[(i % INTERNAL_BUFFER_SIZE)].deliver)
			a_interval_performance->highest_deliverable_inst_id = i;
		else
			break;
	}
}
//------------------------------------------------------------------------------
void execute_phase_2_b(paxos_msg *accept_msg) {
	int i = accept_msg->instance_number;
	int r = accept_msg->round_number;

	if (data[(i % INTERNAL_BUFFER_SIZE)].packet.pack_id != data[(i % INTERNAL_BUFFER_SIZE)].instance_id + 1) {
		printf("process: %d\n", get_config_info()->self_id);
	}
	assert(data[(i % INTERNAL_BUFFER_SIZE)].packet.pack_id == data[(i % INTERNAL_BUFFER_SIZE)].instance_id + 1);

	if (data[(i % INTERNAL_BUFFER_SIZE)].received) {
		if (paxos_instances[(i % INTERNAL_BUFFER_SIZE)].rnd <= r) {
			if (paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vrnd != r) {
				paxos_instances[(i % INTERNAL_BUFFER_SIZE)].rnd = r;
				paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vrnd = r;
				accept_msg->acceptor_count++;
				accept_msg->type = PHASE_2_B;

				if (accept_msg->acceptor_count == get_config_info()->total_num_acceptors)
					if (MULTICASTER)
						paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vval = accept_msg->VALUE;

			} else {
				if (paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vrnd == r) { // This is a hack: we assume the value proposed is the same...
					accept_msg->acceptor_count++;
					accept_msg->type = PHASE_2_B;
					if (accept_msg->acceptor_count == get_config_info()->total_num_acceptors)
						if (MULTICASTER)
							paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vval = accept_msg->VALUE;
				}
			}
		}
	}
	if (!MULTICASTER) {
		unsigned int slen = sizeof(ring_successor_addr);
		int status = sendto(get_config_info()->server_infos[get_config_info()->self_id].successor_sid, accept_msg, sizeof(paxos_msg), 0, (struct sockaddr*) &ring_successor_addr, slen);
		assert(status == sizeof(paxos_msg));
		a_interval_performance->interval_bytes_forward += status;
	}
}
//------------------------------------------------------------------------------
void acceptor_phase_recognizer(int sd, short event, void * arg) {
	int status;
	struct sockaddr_in ring_client_addr;
	unsigned int slen = sizeof(ring_client_addr);

	paxos_msg pm;
	status = recvfrom(sd, &pm, sizeof(paxos_msg), MSG_WAITALL, (struct sockaddr*) &ring_client_addr, &slen);
	assert(status == sizeof(paxos_msg));
	a_interval_performance->interval_bytes_received += status;

	if (status == sizeof(paxos_msg)) {
		a_interval_performance->interval_num_paxos_packets_received++;

		if (pm.type == PHASE_2_B) {
			execute_phase_2_b(&pm);
			if (MULTICASTER) {
				if (pm.acceptor_count == get_config_info()->total_num_acceptors) {
					if (data[(pm.instance_number % INTERNAL_BUFFER_SIZE)].received) {
						learner_deliver();
						multicaster_deliver(&pm);
						if (pm.instance_number > gapEnd)
							gapEnd = pm.instance_number;
						look4highest(a_interval_performance->highest_deliverable_inst_id + 1, gapEnd + 1, pm.instance_number);
					}
				}
			}
		}
	}
}

//----------------------------------------------------------------
void recv_new_packet(int sd, short event, void * arg) {

	UNUSED_ARG(event);
	UNUSED_ARG(arg);

	int status = -1;
	unsigned int len = sizeof(struct sockaddr);

	status = recvfrom(sd, new_data, sizeof(inst), MSG_WAITALL, (struct sockaddr *) ipmAddr, &len);
	assert(status == sizeof(inst));
	a_interval_performance->interval_bytes_multicast += status;

	if (a_interval_performance->highest_deliverable_inst_id != new_data->highestDeliverable) {
		a_interval_performance->highest_deliverable_inst_id = new_data->highestDeliverable;
		learner_deliver();
	}
	if (data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].packet.pack_id == new_data->packet.pack_id) {
		a_interval_performance->interval_num_redundant_packets_received++;
		a_interval_performance->total_num_redundant_packets_received++;
	} else {
		memcpy(&data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)], new_data, sizeof(inst));
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].received = true;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].deliver = false;
		data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].highestDeliverable = -1;
	}
	if (new_data->packet.msgs[0].msg_id != 0) {
		if (COORDINATOR)
			coordinator_execute_phase_2_a(new_data->instance_id, new_data->round_number);
	}
}
//------------------------------------------------------------------------------
